/***************************************************************************************
  *Copyright(C),2010-2016,Sumscope
  *FileName   : Table.cpp
  *Author     : scofined.qi
  *Version    : 1.0
  *Date       : 2016/06/02
  *Desc       : //用于说明此程序文件完成的主要功能
  *Relation   :
  *Other      : //其他内容说明
  *Function   : //主要函数列表，每条记录应包含函数名及功能简要说明
  *History    : //修改历史纪录列表，每条修改记录应包含修改日期，修改者以及修改内容简介
  ***************************************************************************************/

#include "Table.h"
#include "TableSorter.h"
#include "TableCache.h"
#include "TableFile.h"
#include "LevelDBCache.h"

#ifdef WIN32
#define BASSERT assert
#else
#define BASSERT //
#endif

namespace qb{
	namespace ht{
		void Table::WriteListener::OnWrite(int rid, const void* prev, const void* cur, int size, bool insert)
		{
			m_table->UpdateFieldChangedFlags(rid, prev, cur, size, insert);
		}
		Table::Table(const TableDesc& desc, int tableid, int headSize, bool use_ldb) :
			m_tableid(tableid),
			m_desc(desc),
			m_head_size(headSize),
			m_field_change_notify(false),
			m_flag_useldb(use_ldb),
			m_cache(0),
			m_viewsize(4),
			m_increase_size(4)
		{
			m_write_listener = new WriteListener(this);
			memset(m_records, 0, kRecordChannelCount*sizeof(Record));
		}

		Table::~Table()
		{
			// m_records;
			//if (m_record.m_buf){ delete[] m_record.m_buf; }
			for (int i = 0; i < kRecordChannelCount; i++)
			{
				if (m_records[i].m_buf)
					delete[] m_records[i].m_buf;
				m_records[i].m_buf = NULL;
			}
			if (m_cache)		delete m_cache;
			m_cache = NULL;
			if (m_write_listener)	delete m_write_listener;
			m_write_listener = NULL;
		}

		void Table::SetCacheCapcity(int capcity)
		{
			GetCache()->SetCapcity(capcity);
		}
        bool Table::Initialize(const std::string& fname, const std::string& tag,bool temp)
		{
            BASSERT(m_cache==0);
			m_cache_filename = fname;
			m_tagname = tag;
            const std::vector<ColumnInfo>& columns = m_desc.GetColumns();
            BASSERT(columns.size() > 0);
			const ColumnInfo& last = columns.at(columns.size() - 1);
			const ColumnInfo& key_column = columns[m_desc.GetKeyColumn()];
#ifdef USE_LEVELDB
			if (m_flag_useldb)
				m_cache = new LevelDBCache(m_desc.GetRecordSize(), key_column.GetDataType(), key_column.GetFieldBytes());
#endif
			if (m_cache == 0)
				m_cache = new TableCache(m_desc.GetRecordSize(), m_head_size, key_column.GetDataType(), key_column.GetFieldBytes());
			m_cache->SetName(fname);
			m_cache->SetTempFile(temp);
			m_cache->SetViewSize(m_viewsize);
			m_cache->SetIncreaseSize(m_increase_size);
			m_cache->SetTag(tag);
			TableFile* tfile = m_cache->GetFile(); 
			
			m_cache->SetQuotaListener(m_write_listener);
			for (int i = 0; i < kRecordChannelCount; i++)
			{
				m_records[i].m_size = m_cache->GetRecordLength();
				m_records[i].m_buf = new char[m_records[i].m_size + 1];
				m_records[i].m_rid = -1;
				memset(m_records[i].m_buf, 0, m_records[i].m_size + 1);
			}
			if (tfile && tfile->IsInitialized())
				return true;

			return false;
		}
		void Table::Unload()
		{
			if (m_cache)
				m_cache->GetFile()->ReleaseView();
			//if (m_cache)
			//	delete m_cache;
			//m_cache = NULL;
		}
		void Table::Reload()
		{
			//if (m_cache == NULL)
			//{
			//	string fname = m_cache_filename;
			//	string tag = m_tagname;
			//	Initialize(fname, tag);
			//}
		}
		void Table::AddUser(TableUser* user)
		{
			user->SetTable(this);
			m_users[user->GetWindow()] = user;
		}

		/*
		http://tech.it168.com/a2012/1030/1415/000001415603_all.shtml
		参考eXtremeDB的测试性能指标
		硬件配置：Intel(R) Core(TM) i7 CPU 860 @ 2.80GHz，4核8线程， 内存8GB。
		操作系统： Redhat Enterprise Linux 6.0 X64。
		class Record
		{
		uint4 id;
		uint4 i1;
		uint4 i2;
		double d1;
		double d2;
		char<30> s1;
		char<30> s2;
		list;
		unique hash <id> hkey[10000000];
		tree <id> tkey;
		};

		插入测试
		单线程的 1kw	耗时(ms):27286 单条(ms):2.7286 吞吐率(object/s): 366488.3090
		四线程的 1kw	耗时(ms):12873 单条(ms):1.2873 吞吐率(object/s): 776804.6142
		八线程的 1kw	耗时(ms):14197 单条(ms):1.4197 吞吐率(object/s): 704355.5587
		更新测试
		单线程的 1kw	耗时(ms):57740 单条(ms):5.7740 吞吐率(object/s): 173190.200
		四线程的 1kw	耗时(ms):28234 单条(ms):2.8234 吞吐率(object/s): 354173.500
		八线程的 1kw	耗时(ms):28.58 单条(ms):2.8582 吞吐率(object/s): 349870.500
		查询测试
		单线程的 1kw	耗时(ms): 8599 单条(ms):0.8599 吞吐率(object/s):1162925.92
		四线程的 1kw	耗时(ms): 5164 单条(ms):0.5074 吞吐率(object/s):1970540.42
		八线程的 1kw	耗时(ms): 6000 单条(ms):0.5972 吞吐率(object/s):1674235.60
		删除测试
		单线程的 1kw	耗时(ms):28658 单条(ms):2.8658 吞吐率(object/s): 348942.70
		四线程的 1kw	耗时(ms):17741 单条(ms):1.7741 吞吐率(object/s): 563650.20
		八线程的 1kw	耗时(ms):21476 单条(ms):2.1476 吞吐率(object/s): 465622.50
		1:1读写测试
		四线程	71w查询 23更新
		*/
		void Table::ReadRecords(RecordResult& rr)
		{
			int ncount = rr.GetCount();
			for (int i = 0; i < ncount;i++)
			{
				int rid = rr.GetRecordId(i);
				rr.SetRecord(i, ReadRecord(rid, kRecordChannelDefault));
			}
		}
		void Table::SortForUser(int index)
		{
			ScopedLock scoped(m_lock);
			TableUser* user = GetUser(index);
			TableSorter::Get().SortColumns(this, user);
		}

		char* Table::ReadRecordInternal(int rid, kRecordChannel channel)
		{
			m_records[channel].m_rid = rid;
			memset(m_records[channel].m_buf, 0, m_records[channel].m_size);
			if (!m_cache->Read(rid, m_records[channel].m_buf, m_records[channel].m_size))
				return 0;
			return m_records[channel].m_buf;
		}
		const char* Table::ReadRecord(int rid, kRecordChannel channel)
		{
			return ReadRecordInternal(rid, channel);
		}
		const char* Table::ReadRecordDirect(int rid)
		{
			Record& rd = m_records[kRecordChannelDirect];
			rd.m_rid = rid;
			memset(rd.m_buf, 0, rd.m_size);
			m_cache->Read(rid, rd.m_buf, rd.m_size);
			return rd.m_buf;
		}

        void Table::ReadColumns(int rid, const std::vector<kColumn>& columns, char*buf, int size)
		{
			// 读取指定记录的若干列数据到内存buf中，size为这些列的大小之和
			char* ptr = buf;
			const char* record = ReadRecordDirect(rid);
			for (int i = 0; i < (int)columns.size(); i++)
			{
				const ColumnInfo& ci = m_desc.GetColumns().at(columns[i]);
				int bytes = ci.GetFieldBytes();
				memcpy(ptr, record + ci.GetOffset(), bytes);
				ptr += bytes;
			}
            BASSERT(buf + size >= ptr);
		}
		RecordId Table::GetRecordIdByKey(const void* key)
		{
			return GetCache()->GetRecordId((const char*)key);
		}
		RecordId Table::RemoveRecord(const void* key)
		{
			/*
			以key删除一条记录
			如果数据表存在标志列(类型为bool,值为1表示删除,默认0表示不删除),那么
			会将该条记录的标志列字段改为1
			然后将此key从内存的哈希表中删除

			*/
			ScopedLock scoped(m_lock);
			int flagoffset = -1;
			int flagColumn = m_desc.GetFlagColumn();
			if (flagColumn >= 0)
			{
                BASSERT(flagColumn < GetColumnCount());
				const ColumnInfo& flagcol = GetColumnInfo(flagColumn);
                BASSERT(flagcol.GetDataType() == kDataTypeBool||flagcol.GetDataType()==kDataTypeChar);
				flagoffset = flagcol.GetOffset();
                BASSERT(flagoffset < GetRecordLength());
			}
			RecordId rid = GetRecordIdByKey(key);
			if (rid >= 0)
			{
				char* record = ReadRecordInternal(rid);
                BASSERT(record);
				if (record)
				{
					if (flagoffset >= 0)
						record[flagoffset] = (char)((bool)1);
					WriteRecord(record, GetRecordLength());
				}
				m_cache->RemoveKey((const char*)key);
			}
			return rid;
		}
		RecordId Table::WriteRecordDirect(const void* record, int size)
		{
			ScopedLock scoped(m_lock);
            BASSERT(record && size == m_cache->GetRecordLength());
			const char* key = GetKeyColumn(record, size);

			// 检查key
			//int keylen = strlen(key);
			//const ColumnInfo& ci = m_desc.GetColumn(m_desc.GetKeyColumn());
			//BASSERT(keylen > 0 && keylen < ci.GetFieldBytes());
			//for (int i = keylen; i < ci.GetFieldBytes();i++){ BASSERT(key[i] == 0); }

			RecordId existid = m_cache->GetRecordId(key);
			RecordId rid = m_cache->Write(existid, record, size, key,true);
			return rid;
		}
		void Table::WriteRecordDirect(int rid, const void* record, int size)
		{
            BASSERT(rid >= 0);
			m_cache->Write(rid, record, size, nullptr, true);
		}
		RecordId Table::WriteRecord(const void* record, int size)
		{
			ScopedLock scoped(m_lock);
            BASSERT(record && size == m_cache->GetRecordLength());

			// get key from record with key columns
			//RecordKey key;
			//GetKeyColumn(record, size, key);
			const char* key = GetKeyColumn(record, size);
			RecordId existid = -1;
			{
				existid = m_cache->GetRecordId(key);
			}
			

			Record& olddata = m_records[kRecordChannelOldData];
			if (existid < 0)
			{
				memset(olddata.m_buf, 0, size);
			}
			else
				m_cache->Read(existid, olddata.m_buf, size);


			// 要求key是其中一个列
			RecordId rid = -1;
			{
				rid = m_cache->Write(existid, record, size, key);
			}
			

			// 字段变化m_cache 无法得知
			if (m_field_change_notify && m_users.size()>0)
			{
				UpdateFieldChangedFlags(rid, olddata.m_buf, record, size, rid != existid);
			}
				
			{
				//STACK_METRICS("TableUser::InsertRow");
				for (TableUserMap::iterator it = m_users.begin(); it != m_users.end(); ++it)
				{
					//CMETRICS("user insert row");
					TableUser* user = it->second;
					user->InsertRow(this, rid, record, size, m_field_changed, rid != existid);
				}
			}
			return rid;
			
		}
        void Table::WriteRecordFields(const void* record, int size,int rid,const std::vector<bool>& fields)
		{
            BASSERT(rid >= 0);
			ReadRecord(rid, kRecordChannelDefault);
			Record& rd = m_records[kRecordChannelDefault];
			for (int i = 0; i < (int)fields.size();i++)
			{
				if (!fields[i])	continue;
				const ColumnInfo& ci = GetColumnInfo(i);
				int offset = ci.GetOffset();
				memcpy(rd.m_buf + offset, (const char*)record + offset, ci.GetFieldBytes());
			}
			WriteRecord(rd.m_buf, rd.m_size);
		}
		void Table::UpdateFieldChangedFlags(int rid, const void* prev, const void* cur, int size, bool insert)
		{
			// 开启后从 45w-->36w
			// 对比前后两个记录，看那些字段变化了
			// 着重优化
			m_field_changed.clear();
			int count = m_desc.GetColumns().size();
			for (int col = 0; col < count; col++)
			{
				const ColumnInfo& ci = m_desc.GetColumn(col);
				bool changed = insert ? true : (memcmp((const char*)prev + ci.GetOffset(), (const char*)cur + ci.GetOffset(), ci.GetFieldBytes()) != 0);
				m_field_changed.push_back(changed);
			}

		}


        TableUser* Table::GetUser(void* wnd)
		{
			TableUserMap::iterator it = m_users.find(wnd);;
			return it != m_users.end() ? it->second : NULL;
		}

		TableUser* Table::GetUser(int index)
		{
			int i = 0;
			for (TableUserMap::iterator it = m_users.begin(); it != m_users.end(); ++it, i++)
			{
				if (i == index)
					return it->second;
			}
			return NULL;
		}

		int Table::GetRecordLength() const
		{
			return m_cache ? m_cache->GetRecordLength() : 0;
		}

		ITableCache* Table::GetCache()
		{
            BASSERT(m_cache);
			return m_cache;
		}

		int Table::GetRemapCount() const
		{
			if (m_cache && m_cache->GetFile())
				return m_cache->GetFile()->GetRemapCount();
			return 0;
		}

		int Table::GetMaxRecordId() const
		{
			return m_cache->GetMaxRecordId();
		}

		//void Table::GetKeyColumn(const void* record, int size, RecordKey& key)
		//{
		//	BASSERT(m_cache->GetRecordLength() == size && record);
		//	const ColumnInfo& ci = m_desc.GetColumn(m_desc.GetKeyColumn());
		//	// 后续需要针对每列有一个解码器
		//	// 此处假设key column为 chararray
		//	key = RecordKey((const char*)record + ci.GetOffset());
		//	BASSERT(!key.empty());
		//}
		const char* Table::GetKeyColumn(const void* record, int size)
		{
            BASSERT(m_cache->GetRecordLength() == size && record);
            const ColumnInfo& ci = m_desc.GetColumn(m_desc.GetKeyColumn());
			// 后续需要针对每列有一个解码器
			// 此处假设key column为 chararray/nString
			if (ci.GetDataType() == kDataTypeNString)
				return (const char*)record + ci.GetOffset() + sizeof(unsigned short) * 2;

            BASSERT(ci.GetDataType() == kDataTypeCharArray);
			return (const char*)record + ci.GetOffset();
		}
		kColumn Table::GetColumnByName(const char* name)
		{
			if (m_column_name_map.size() != m_desc.GetColumnCount())
			{
				m_column_name_map.clear();
				for (int i = 0; i < (int)m_desc.GetColumnCount();i++)
				{
					const ColumnInfo& ci = GetColumnInfo(i);
					m_column_name_map.insert(ColumnNameMap::value_type(ci.GetField(),i));
				}
			}
			if (name)
			{
				ColumnNameMap::iterator it = m_column_name_map.find(name);
				if (it != m_column_name_map.end())
					return (kColumn)it->second;
			}
			return (kColumn) (- 1);
		}

		kColumn Table::GetKeyColumn()
		{
			return m_desc.GetKeyColumn();
		}

		int Table::GetRecordCount() const
		{
			return m_cache->GetRecordCount();
		}

		ViewInfo Table::GetViewInfo()
		{
			if (m_cache && m_cache->GetFile())
				return m_cache->GetFile()->GetViewInfo();
			return ViewInfo();
		}
		int Table::GetProperty(kTableProperty tp)
		{
			TableFile* file = m_cache ? m_cache->GetFile() : 0;
			if (m_cache == 0 )
				return 0;
			switch (tp)
			{
			case kTablePropertyMemory:
			{
				int bytes = sizeof(*this);
				bytes += m_cache ? m_cache->GetMemoryBytes() : 0;
				bytes += kRecordChannelCount*m_desc.GetRecordSize();
				return bytes;
			}
			case kTablePropertyFileSize:return file ? file->GetFileSize():0; break;
			case kTablePropertyFileViewSize:return file ? file->GetViewSize():0; break;
			case kTablePropertyFilePageSize:return file ? file->GetPageSize():0; break;
			case kTablePropertyCacheSwapCount:m_cache->GetSwapCount(); break;
			case kTablePropertyCacheWriteCount:m_cache->GetWriteCount(); break;
			case kTablePropertyRecordSize:return m_desc.GetRecordSize(); break;
			case kTablePropertyKeySize:return m_cache->GetKeyBytes(); break;
			case kTablePropertyQuotaCapcity:return m_cache->GetQuotaCapcity(); break;
			case kTablePropertyQuotaLength:return m_cache->GetQuotaLength(); break;
			default:break;
			}
			return 0;
		}
		std::string Table::GetCacheFileName()
		{
			return m_cache->GetName();
		}
		////////////////////////////////////////////////////////////////////////////////
		class TablePropertyNames{
			typedef std::map<std::string, kTableProperty> PropertyNameMap;
		public:
			PropertyNameMap		m_properties;
			static TablePropertyNames& Get(){ 
				static TablePropertyNames  tpn; 
				return tpn; }
			TablePropertyNames()
			{
				m_properties["memory"] = kTablePropertyMemory;
				m_properties["file_size"] = kTablePropertyFileSize;
				m_properties["file_view_size"] = kTablePropertyFileViewSize;
				m_properties["file_page_size"] = kTablePropertyFilePageSize;
				m_properties["cache_swap_count"] = kTablePropertyCacheSwapCount;
				m_properties["cache_write_count"] = kTablePropertyCacheWriteCount;
				m_properties["cache_quota_capcity"] = kTablePropertyQuotaCapcity;
				m_properties["cache_quota_length"] = kTablePropertyQuotaLength;
				m_properties["record_size"] = kTablePropertyRecordSize;
				m_properties["key_size"] = kTablePropertyKeySize;
			}
			kTableProperty GetPropertyByName(const char* name)
			{
				if (name && strlen(name) > 0)
				{
					PropertyNameMap::iterator it = m_properties.find(name);
					return it != m_properties.end() ? it->second : kTablePropertyNone;
				}
				return kTablePropertyNone;
			}
			const char*    GetPropertyName(kTableProperty tp)
			{
				if (tp > kTablePropertyNone && tp < kTablePropertyMax)
				{
					PropertyNameMap::iterator it = m_properties.begin();
					for (; it != m_properties.end();++it)
					{
						if (it->second == tp)
							return it->first.c_str();
					}
				}
				static const char* blank = "";
				return blank;
			}
		};
		kTableProperty Table::GetPropertyByName(const char* name)
		{
			return TablePropertyNames::Get().GetPropertyByName(name);
		}
        const char* Table::GetPropertyName(kTableProperty tp)
		{
			return TablePropertyNames::Get().GetPropertyName(tp);
		}
		void Table::DumpStatus(int mode)
		{
            const std::vector<ColumnInfo>& columns = m_desc.GetColumns();
			const ColumnInfo& key_column = columns[m_desc.GetKeyColumn()];
			int keysize = key_column.GetFieldBytes();
			int bytes = m_cache ? m_cache->GetMemoryBytes() : 0;
			int ncount = m_cache ? m_cache->GetRecordCount() : 0;
		}
	}//namespace ht
}//namespace qb
